package com.szwistar.common.comm.jsonrpc;

import com.szwistar.common.ERR;
import com.szwistar.common.comm.AsyncResult;
import com.szwistar.common.comm.jsonrpc.http.WebController4JsonRpc;
import com.szwistar.common.datastruct.JsonMapper;
import com.szwistar.common.datastruct.ObjectMethod;
import com.szwistar.common.datastruct.Reflector;
import com.szwistar.common.datastruct.Utils;
import com.szwistar.common.os.TimerEx;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public abstract class JsonRpcEndpointAbstract implements JsonRpcEndpoint {
	protected static final Logger LOG = LoggerFactory.getLogger(JsonRpcEndpointAbstract.class);
	protected static final Logger LOGCOMM = LoggerFactory.getLogger("comm");

	/**
	 * 检查主题订阅过滤器是否合法
	 * <h1>JsonRpc 消息订阅遵守与 MQTT 相同的主题匹配规则:</h1>
	 * <li>单层通配符  +: 单层通配符只能匹配一层主题，可以通配主题的某一层。如：event/+/turnOn 可以匹配 event/123/turnOn 和 event/456/turnOn
	 * <li>多层通配符  #: 多层通配符可以匹配于多层主题，必须放在最后。如：event/# 可以匹配 event/123/turnOn 和 event/456/turnOn
	 * <li>+/# 可以搭配使用，如：ehome/gateway4/mdm/+/3E00FFFF00000002/# 可以匹配 3E00FFFF00000002 的所有调用和通知消息
	 * @param filter 不能为空，不能以“/”开头；如果有“#”号，则必须只能有一个，且必须在最后
	 */
	protected static boolean validateFilter(String filter) {
		// 不能为空，不能以“/”开头
		if(Utils.isEmpty(filter) || filter.indexOf('/') == 0) { return false; }
		// 如果有“#”号，则必须只能有一个，且必须在最后
		int pp = filter.indexOf('#');
		return (pp == -1) ? true : (pp == filter.lastIndexOf('#') && pp == filter.length() - 1);
	}

	/**
	 * 检查主题是否合法
	 * @param topic 必须是明确的主题，不能为空，不能以“/”开头，不能包含通配符
	 */
	protected static boolean validateTopic(String topic) {
		return !(Utils.isEmpty(topic) || topic.indexOf('/') == 0 || topic.indexOf('+') >= 0 || topic.indexOf('#') >= 0);
	}

	/**
	 * 判断一个主题订阅过滤器是否包含指定的主题
	 * @param filter 主题订阅过滤器
	 * @param topic 要发送的主题(必须是明确的主题，不能包含通配符)
	 * @return 如果 filter 包含要发送的主题，则返回 true；否则返回 false
	 */
	protected static boolean filterTopic(String filter, String topic) {
		if(!validateFilter(filter) || !validateTopic(topic)) { return false; }

		// 层次分解
		String[] fParts = filter.split("/");
		String[] tParts = topic.split("/");

		// 对主题的逐个层次进行比较
		for(int i=0; i<tParts.length; i++) {
			// 通配或相等，则本层次匹配，继续匹配下一层次
			if(fParts[i].equals("+")) { continue; }
			// 如果后面的都通配，则匹配
			if(fParts[i].equals("#")) { return true; }
			// 如果不严格相等，则不匹配
			if(!fParts[i].equals(tParts[i])) { return false; }
		}

		return true;
	}

	/**
	 * 判断一组主题订阅过滤器是否包含指定的主题
	 * @param filters 主题订阅过滤器
	 * @param topic 要发送的主题
	 * @return 如果 filters 包含要发送的主题，则返回 true；否则返回 false
	 */
	protected static boolean filterTopic(List<String> filters, String topic) {
		if(Utils.isEmpty(filters) || topic == null) { return false; }
		for(String filter: filters) {
			if(filterTopic(filter, topic)) { return true; }
		}
		return false;
	}

	// 本端自己的监听 Topic，对于 P2p 类连接不起作用，对 MQTT 连接有效
	protected String myLocalTopic;
	// ping 平均延时
	protected int pingDelayMs = 0;
	// 处于 pending 状态的 return 消息的字典
	protected final Map<Long, AsyncResult<JsonRpcMessageReturn>> allPending = new ConcurrentHashMap<>();
	// 提供的 rpc 服务函数表
	protected Map<String, ObjectMethod> methodsMap = new HashMap<String, ObjectMethod>();
	// 提供的 notify 服务函数表
	protected Map<String, ObjectMethod> eventsMap = new HashMap<String, ObjectMethod>();

	@Override
	public int getDelay(String target) { return pingDelayMs; }

	public JsonRpcEndpointAbstract() {
		super();
	}

	@Override
	public void registerService(String svcname, Object obj) {
		List<Method> methods;

		// 登记所有 RPC 服务函数
		methods = Reflector.getMethods(obj.getClass(), null, null, JsonRpcMessageReturn.class, JsonRpcMessageCall.class);
		for(Method method: methods) {
			// 服务中的服务函数调用名为："服务名称.函数名"，如： "xiaodu.discover"
			String name = svcname + "." + method.getName();
			//LOG.debug("登记 RPC 函数：{} => {}.{}()", name, service.getClass().getName(), method.getName());
			methodsMap.put(name, new ObjectMethod(obj, method));
		}

		// 登记所有事件处理函数
		// 服务中的服务函数调用名为："服务名称.函数名"，如： "xiaodu.discover"
		methods = Reflector.getMethods(obj.getClass(), null, null, void.class, JsonRpcMessageNotify.class);
		for(Method method: methods) {
			// 服务中的服务函数调用名为："服务名称.函数名"，如： "xiaodu.discover"
			String name = svcname + "." + method.getName();
			//LOG.debug("登记事件处理函数：{} => {}.{}()", name, service.getClass().getName(), method.getName());
			eventsMap.put(name, new ObjectMethod(obj, method));
		}
	}

	@Override
	public void registerService(JsonRpcService service) {
		registerService(service.getName(), service);
	}

	/**
	 * TX: 派生类需实现此函数，发送消息到指定的外部地址
	 * @param target 目标地址,对于不同的承载层协议,目标地址的定义不同
	 * @param msg 要发送的消息
	 * @param qos 消息优先级
	 */
	public abstract void send(String target, JsonRpcMessage msg, int qos) throws Exception;

	/**
	 * RX: 派生类需要调用此函数,提供调用者发来的消息,以便进行处理
	 */
	@Override
	public void onMessageReceived(JsonRpcMessage msg) {
		// 标记消息的接收接口（如果已经标记了的话，就不要覆盖它了，可能是派生类标记过了）
		if(msg.getEndpoint() == null) { msg.setEndpoint(this); }

		try {
			// Case1: RPC 请求
			if(msg instanceof JsonRpcMessageCall) {
				final JsonRpcMessageCall request = (JsonRpcMessageCall)msg;

				if(this instanceof WebController4JsonRpc) {
					// 如果是 JsonRpc@HTTP，则必须在本线程中处理，否则会导致发回响应时，HTTP 连接已经断开
					handleRequest(request);
				} else {
					// 不直接在此线程中处理此消息了(因为可能阻塞)，在一个线程池中处理，以免影响消息的收发
					TimerEx.runLater(() -> handleRequest(request));
				}
			}
			// Case2: RPC 请求的响应
			else if(msg instanceof JsonRpcMessageReturn) {
	            // 在未返回的消息中匹配
	            AsyncResult<JsonRpcMessageReturn> pendingReturn = allPending.get(msg.getMsgId());
	            if(pendingReturn == null) {
	            	LOGCOMM.warn("未知来源的消息: {}", msg.getMsgId());
	            	return;
	            }
	            // 异步调用完成
	            pendingReturn.doneWithResult(0, null, (JsonRpcMessageReturn) msg);
	            // 从待返回队列中移除
	            allPending.remove(msg.getMsgId());
			}
			// Case3: 事件通知处理
			else if(msg instanceof JsonRpcMessageNotify) {
				final JsonRpcMessageNotify notify = (JsonRpcMessageNotify)msg;
				// 不直接在此线程中处理此消息了(因为可能阻塞)，在一个线程池中处理，以免影响消息的收发
				//handleNotify(notify);
				TimerEx.runLater(() -> handleNotify(notify));
			}
			// Case4: PING 请求
			else if(msg instanceof JsonRpcMessagePing) {
				JsonRpcMessagePing request = (JsonRpcMessagePing)msg;
				// 注意：这里的 QOS 一定要设置成 1 ？！否则会导致发送阻塞？！！！
				send(request.getFrom(), new JsonRpcMessagePingAck(request), 0, 1);
			}
			// Case5: PING 响应
			else if(msg instanceof JsonRpcMessagePingAck) {
				JsonRpcMessagePingAck ack = (JsonRpcMessagePingAck)msg;
				pingDelayMs = ack.getDelay();
				LOGCOMM.debug("ping() 响应时间 {} ms", pingDelayMs);
			}
			// Case: 其它
			else {
				LOGCOMM.warn("不支持的 JsonRpc 消息: {}", msg);
			}
		} catch (Exception e) {
			LOGCOMM.error("JsonRpc 接收消息处理失败！", e);
		}
	}

	protected void handleRequest(JsonRpcMessageCall request) {
		// 处理请求
		JsonRpcMessage retMsg = null;

		do {
			// 处理订阅请求
			// 当请求方法为 "_subscribe" 时，默认为消息订阅请求
			if(request.getMethod().equals("_subscribe")) {
				retMsg = handleSubscribe(request);
				break;
			}

			// 查找 RPC 方法
			ObjectMethod service = methodsMap.get(request.getMethod());
			if(service == null) {
				LOGCOMM.error("RPC 请求调用的方法 {} 不存在！", request.getMethod());
				retMsg = JsonRpcMessageReturn.error(request, ERR.COMMAND, "RPC 请求调用的方法不存在！");
				break;
			}

			try {
				// 调用 RPC 方法
				LOGCOMM.debug("处理 RPC 请求：{}", request);
				Object ret = service.getMethod().invoke(service.getOwner(), request);

				// 处理返回值
				if(ret instanceof JsonRpcMessage) {
					retMsg = (JsonRpcMessage) ret;
				} else {
					retMsg = JsonRpcMessageReturn.error(request, ERR.SYS, "RPC 函数返回非法值！");
				}
			} catch(InvocationTargetException e) {
				LOGCOMM.error("RPC 调用失败！{}", e.getTargetException().getCause().getMessage());
				retMsg = JsonRpcMessageReturn.error(request, ERR.SYS, "调用失败！" + e.getTargetException().getCause().getMessage());
			} catch(Exception e) {
				LOGCOMM.error("RPC 调用失败！", e);
				retMsg = JsonRpcMessageReturn.error(request, ERR.SYS, "调用失败！" + e.getMessage());
			}
		} while(false);

		// 发送调用返回消息
		send(request.getFrom(), retMsg, 0, 1);
	}

	protected void handleNotify(JsonRpcMessageNotify notify) {
		try {
			// 查找事件处理方法
			ObjectMethod service = eventsMap.get(notify.getMethod());
			if(service == null) {
				LOGCOMM.error("无对应的事件处理函数！{}", notify);
				return;
			}

			// 调用事件处理方法
			LOGCOMM.debug("调用事件处理器 {} 处理事件通知：\n{}", service, notify);
			service.getMethod().invoke(service.getOwner(), notify);
		} catch(Exception e) {
			LOGCOMM.error("事件通知 {} 处理失败！", notify, e);
		}
	}

	@Override
	public JsonRpcMessage send(String target, JsonRpcMessage msg, int timeout, int qos) {
		try {
			AsyncResult<JsonRpcMessageReturn> pendingReturn = null;

			if(timeout > 0) {
				// 注意：为确保返回消息不会在把消息放入待返回消息队列之前就到达，要在 send() 之前就放入
				pendingReturn = new AsyncResult<JsonRpcMessageReturn>(timeout);
				allPending.put(msg.getMsgId(), pendingReturn);
			}

	    	// 发送消息
			//if(!(msg instanceof JsonRpcMessagePingAck)) { LOGCOMM.debug("发送消息 {} 到 {}", msg, target); }
			send(target, msg, qos);

			if(timeout > 0) {
				// 等待对方返回
				JsonRpcMessageReturn ret = pendingReturn.get();
				if(ret == null) { return JsonRpcMessageReturn.error(msg, ERR.TIMEOUT, "等待返回超时！msgId = " + msg.getMsgId()); }
				if(msg instanceof JsonRpcMessageCall) {
					LOGCOMM.debug("调用耗时: {} ms", System.currentTimeMillis() - ((JsonRpcMessageCall)msg).getMs());
				}
				return ret;
			}

			return JsonRpcMessageReturn.ok(msg, null);
		}
		catch (Exception e) {
			LOGCOMM.error("JsonRpc 发送消息失败！{}\n{}", msg, e.getMessage());
			return JsonRpcMessageReturn.error(msg, ERR.SYS, e.toString());
		}
		finally {
			if(timeout > 0) {
				// 不管是否正常返回，都要清除等待队列中此次异步调用的结果
				allPending.remove(msg.getMsgId());
			}
		}
	}

	public static String packParam(Object obj) throws Exception {
		if(obj == null) { return null; }
		return (obj instanceof String) ? (String) obj : JsonMapper.toJson(obj);
	}

	@Override
	public JsonRpcMessageReturn rpc(String target, String token, String method, Object params, int timeout) {
		try {
			JsonRpcMessage msg = new JsonRpcMessageCall(myLocalTopic, token, method, packParam(params));
			return (JsonRpcMessageReturn) send(target, msg, timeout, 2);
		} catch (Exception e) {
			LOG.error("参数编码失败！{}", e.getMessage());
			return JsonRpcMessageReturn.error(null, ERR.BAD_PARAM, "参数编码失败！" + e.getMessage());
		}
	}

	@Override
	public void notify(String targetTopic, String token, String name, Object params) {
		try {
			JsonRpcMessage msg = new JsonRpcMessageNotify(myLocalTopic, token, name, packParam(params));
			LOGCOMM.debug("向 '{}' 发送主题为 '{}' 的通知：{}", this, targetTopic, msg);
			send(targetTopic, msg, 0, 1);
		} catch (Exception e) {
			LOG.error("参数编码失败！{}", e.getMessage());
		}
	}

	@Override
	public JsonRpcMessageReturn subscribe(String target, String token, String[] add, String[] del, int timeout) {
		Map<String, String[]> params = new HashMap<>();
		params.put("add", add);
		params.put("del", del);
		return rpc(target, token, "_subscribe", params, timeout);
	}

	@Override
	public void ping(String targetTopic) {
		JsonRpcMessage msg = new JsonRpcMessagePing(myLocalTopic);
		send(targetTopic, msg, 0, 1);
	}

	// for test only
	public static void main(String[] args) {
		System.out.println(filterTopic("/#", "a/b/c")); // false
		System.out.println(filterTopic("#", "/a/b/c")); // false
		System.out.println(filterTopic("#", "a/+/c"));  // false
		System.out.println(filterTopic("#", "a/b/c"));  // true
		System.out.println(filterTopic("a/+/c", "a/b/c"));  // true
		System.out.println(filterTopic("a/+/c/#", "a/b/c/d/e/f"));  // true
		System.out.println(filterTopic("a/+/c/+/e/#", "a/b/c/d/e/f"));  // true
	}
}
